package edu.nepu.flink.api.aggregation;

import edu.nepu.flink.api.bean.WaterSensor;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @Date 2024/2/28 21:48
 * @Created by chenshuaijun
 */
public class RichFunction {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        DataStreamSource<WaterSensor> elementSource = env.fromElements(
                new WaterSensor("s1", 1L, 7),
                new WaterSensor("s1", 3L, 9),
                new WaterSensor("s2", 2L, 8),
                new WaterSensor("s2", 4L, 10),
                new WaterSensor("s3", 5L, 11)
        );

        /**
         * 每一个算子都有一个与之对应的富函数接口。
         * 例如map算子是可以有两种实现的：（1）MapFunction（2）RichMapFunction
         * RichMapFunction相比于普通的MapFunction他多了对于上下文的信息
         */

        elementSource.map(new RichMapFunction<WaterSensor, String>() {

            // 会在最开始初始化的时候执行的方法
            @Override
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);
                // 这个里面我们可以获取程序运行的上下文信息 =
                // 拿到子任务的编号
                System.out.println(getRuntimeContext().getIndexOfThisSubtask());
                // 获得作业的id
                System.out.println(getRuntimeContext().getJobId());
                // 还有状态的创建...都可以使用getRuntimeContext()来实现
            }

            // 主要的处理逻辑
            @Override
            public String map(WaterSensor value) throws Exception {
                return null;
            }

            // 在程序正常退出的时候这部分的逻辑是会被执行的
            @Override
            public void close() throws Exception {
                super.close();
            }
        });

        env.execute();
    }
}
